﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using AzureMRCore.DataCommunication;
using AzureMRCore.DataTypes;
using AzureMRCore.InputFormat;

namespace AzureMRCore.Primitives
{
    public class BCastServiceReceiver
    {
        private static BCastServiceReceiver _receiver;

        private ConcurrentDictionary<string, ConcurrentDictionary<int, BCastDataProduct>> _fetchingBCastData;

        private readonly ConcurrentDictionary<string, IInputFormat> _inMemCache;
        //  var fetchedBCastData = new ConcurrentDictionary<int, BCastDataProduct>();
        private BCastServiceReceiver()
        {
            _fetchingBCastData = new ConcurrentDictionary<string, ConcurrentDictionary<int, BCastDataProduct>>();
            _inMemCache = new ConcurrentDictionary<string, IInputFormat>();
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public static BCastServiceReceiver GetInstance()
        {
            return _receiver ?? (_receiver = new BCastServiceReceiver());
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public void AddAllGatherData(DataItem dataItem)
        {
            string key = dataItem.JobID + "_" + dataItem.Iteration;
            var bCastDataProducts = _fetchingBCastData.GetOrAdd(key,
                                                                data =>
                                                                new ConcurrentDictionary<int, BCastDataProduct>());
            Type inputFormatType = Type.GetType(dataItem.InputFormatType);
            var inputFormat = (IInputFormat) Activator.CreateInstance(inputFormatType);
            bCastDataProducts.TryAdd(dataItem.MapIndex, new BCastDataProduct
                                                            {
                                                                IsSingle = true,
                                                                Key = 0,
                                                                Value = inputFormat.FromBytes(dataItem.Data)
                                                            });
            if (bCastDataProducts.Count == dataItem.NumMapTasks)
            {
                inputFormat.Init(bCastDataProducts, dataItem.NumMapTasks);
                _inMemCache.TryAdd(key, inputFormat);
                ConcurrentDictionary<int, BCastDataProduct> task;
                _fetchingBCastData.TryRemove(key, out task);

                // trying to remove old broadcast data from memcache.
                string oldKey = dataItem.JobID + "_" + (dataItem.Iteration - 1);
                IInputFormat temp;
                _inMemCache.TryRemove(oldKey, out temp);
            }
        }

        public void AddSimpleBCastData(DataItem dataItem)
        {
            string key = dataItem.JobID + "_" + dataItem.Iteration;

            Type inputFormatType = Type.GetType(dataItem.InputFormatType);
            var inputFormat = (IInputFormat) Activator.CreateInstance(inputFormatType);

            inputFormat.FromBytes(dataItem.Data);
            _inMemCache.TryAdd(key, inputFormat);

            // trying to remove old broadcast data from memcache.
            string oldKey = dataItem.JobID + "_" + (dataItem.Iteration - 1);
            IInputFormat temp;
            _inMemCache.TryRemove(oldKey, out temp);
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public bool ContainsBCastData(string tableName)
        {
            return _inMemCache.ContainsKey(tableName);
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public IInputFormat GetBCastData(string tableName)
        {
            var item = _inMemCache[tableName];
            return item;
        }
    }

    public class BCastDataProduct
    {
        public object Key { get; set; }
        public IValue Value { get; set; }
        public bool IsSingle { get; set; }
    }
}